﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Apache.Cassandra;

namespace Sandy
{
    public class MutationTracker
    {
        private readonly SandyClient _client;

        private Dictionary<string, Dictionary<string, List<Mutation>>> _mutationMap;

        internal MutationTracker(SandyClient client)
        {
            _client = client;
            _mutationMap = new Dictionary<string, Dictionary<string, List<Mutation>>>();
        }

        public void Insert(string familyName, string key, string columnName, string value)
        {
            var family = _client.GetFamily(familyName);

            // Create the column
            Column col = new Column
            {
                Name = family.ColumnNameToBytes(columnName),
                Value = family.GetValueBytes(value),
                Timestamp = DateTimeOffset.UtcNow.UtcTicks
            };

            var colOrSuperCol = new ColumnOrSuperColumn { Column = col };

            // Add to the map
            AddMutation(familyName, key, new Mutation { Column_or_supercolumn = colOrSuperCol });
        }

        public void Insert(string familyName, string key, string superCol, string column, string value)
        {
            var family = _client.GetFamily(familyName);

            // Create the column
            Column col = new Column
            {
                Name = family.ColumnNameToBytes(column),
                Value = family.GetValueBytes(value),
                Timestamp = DateTimeOffset.UtcNow.UtcTicks
            };

            // Create the supercol object
            var superColObj = new SuperColumn
            {
                Name = family.SuperColumnNameToBytes(superCol),
                Columns = new List<Column>() { col }
            };

            var colOrSuperCol = new ColumnOrSuperColumn { Super_column = superColObj };

            // Add to the map
            AddMutation(familyName, key, new Mutation { Column_or_supercolumn = colOrSuperCol });
        }

        /// <summary>
        /// Delete an entire key.
        /// </summary>
        /// <param name="familyName"></param>
        /// <param name="key"></param>
        public void Delete(string familyName, string key)
        {
            var del = new Deletion
                          {
                              // Predicate = new SlicePredicate(),
                              Timestamp = DateTimeOffset.UtcNow.UtcTicks
                          };

            AddMutation(familyName, key, new Mutation { Deletion = del });
        }

        /// <summary>
        /// Delete several keys
        /// </summary>
        /// <param name="familyName">The family for each key</param>
        /// <param name="keys">The keys to delete</param>
        public void Delete(string familyName, IEnumerable<string> keys)
        {
            var del = new Deletion
            {
                // Predicate = new SlicePredicate(),
                Timestamp = DateTimeOffset.UtcNow.UtcTicks
            };

            foreach (string key in keys)
            {
                AddMutation(familyName, key, new Mutation { Deletion = del });
            }
        }

        /// <summary>
        /// Delete several keys
        /// </summary>
        /// <param name="familyName">The family for each key</param>
        /// <param name="keys">The keys to delete</param>
        public void Delete(string familyName, params string[] keys)
        {
            Delete(familyName, (IEnumerable<string>)keys);
        }

        /// <summary>
        /// Delete a single column
        /// </summary>
        /// <param name="familyName"></param>
        /// <param name="key"></param>
        /// <param name="column"></param>
        public void DeleteColumn(string familyName, string key, string column)
        {
            var family = _client.GetFamily(familyName);
            var slicePredicate = new SlicePredicate()
            {
                Column_names = new List<byte[]>()
                 {
                       family.ColumnNameToBytes(column)
                 }
            };

            var del = new Deletion
            {
                Timestamp = DateTimeOffset.UtcNow.UtcTicks,
                Predicate = slicePredicate
            };

            AddMutation(familyName, key, new Mutation { Deletion = del });
        }

        /// <summary>
        /// Deletes a number of columns
        /// </summary>
        /// <param name="familyName"></param>
        /// <param name="key"></param>
        /// <param name="columns"></param>
        public void DeleteColumns(string familyName, string key, params string[] columns)
        {
            var family = _client.GetFamily(familyName);
            var slicePredicate = new SlicePredicate
                                     {
                                         Column_names = columns.Select(family.ColumnNameToBytes).ToList()
                                     };

            var del = new Deletion
                          {
                              Timestamp = DateTimeOffset.UtcNow.UtcTicks,
                              Predicate = slicePredicate
                          };

            AddMutation(familyName, key, new Mutation { Deletion = del });
        }

        /// <summary>
        /// Delete a super column for a given key
        /// </summary>
        /// <param name="familyName"></param>
        /// <param name="key"></param>
        /// <param name="superColumn"></param>
        public void DeleteSuperColumn(string familyName, string key, string superColumn)
        {
            var del = new Deletion
                          {
                              Timestamp = DateTimeOffset.UtcNow.UtcTicks,
                              Super_column = _client.GetFamily(familyName).SuperColumnNameToBytes(superColumn)
                          };

            AddMutation(familyName, key, new Mutation { Deletion = del });
        }

        /// <summary>
        /// Delete a single column within a supercolumn
        /// </summary>
        /// <param name="familyName"></param>
        /// <param name="key"></param>
        /// <param name="superColumn"></param>
        /// <param name="column"></param>
        public void DeleteSubcolumn(string familyName, string key, string superColumn, string column)
        {
            var family = _client.GetFamily(familyName);
            var slicePredicate = new SlicePredicate()
                                     {
                                         Column_names = new List<byte[]>()
                                                            {
                                                                family.ColumnNameToBytes(column)
                                                            }
                                     };

            var del = new Deletion
                          {
                              Timestamp = DateTimeOffset.UtcNow.UtcTicks,
                              Super_column = family.SuperColumnNameToBytes(superColumn),
                              Predicate = slicePredicate
                          };

            AddMutation(familyName, key, new Mutation { Deletion = del });
        }


        private void AddMutation(string familyName, string key, Mutation m)
        {
            Dictionary<string, List<Mutation>> keyMutations;
            if (!_mutationMap.TryGetValue(key, out keyMutations))
            {
                // Create a new mutation dict for this key
                keyMutations = new Dictionary<string, List<Mutation>>();

                // Add the mutation 
                keyMutations.Add(familyName, new List<Mutation>() { m });

                // Add back to the main map
                _mutationMap.Add(key, keyMutations);

                return;
            }

            // This key already has mutations
            List<Mutation> mutationList;
            if (!keyMutations.TryGetValue(familyName, out mutationList))
            {
                // Create a new mutation list for this key, add the mutation
                keyMutations.Add(familyName, new List<Mutation>() { m });

                return;
            }

            // This key already has mutations in this column family
            mutationList.Add(m);
        }

        public void SaveChanges()
        {
            // Perform batch_mutate
            _client.Client.batch_mutate(_client.Keyspace, _mutationMap, ConsistencyLevel.ONE);
            _mutationMap = new Dictionary<string, Dictionary<string, List<Mutation>>>();
        }

        ~MutationTracker()
        {
            if (_mutationMap.Count != 0)
            {
                throw new MutationsRemainingException("Mutations remaining in tracker.");
            }
        }
    }

    public class MutationsRemainingException : Exception
    {
        public MutationsRemainingException(string message)
            : base(message)
        {
            // 
        }
    }
}
